#!/usr/bin/env python
#
# Copyright 2009 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""Tests for the FUSE filesystem for App Engine.

NOTE: Must be run as root to do the crazy chmod/chowning required for testing.

These are blackbox tests that exercise the various parts of the server-side
and client/server interaction through filesystem operations.

To run without verbose server logging, pipe stderr to NULL:
  ./aefs_test.py 2>/dev/null
"""

__author__ = 'bslatkin@gmail.com (Brett Slatkin)'

import logging
logging.basicConfig(
    level=logging.DEBUG, format='Test %(levelname)-8s] %(message)s')

import os
import re
import signal
import socket
import stat
import subprocess
import sys
import tempfile
import time
import unittest

import aefs

################################################################################
# Test config

def guess_sdk_path():
  guess = subprocess.Popen(
      'dirname $(readlink $(which dev_appserver.py))',
      stdout=subprocess.PIPE,
      shell=True).communicate()[0].strip()
  logging.info('Using guessed SDK path: %s', guess)
  return guess

SDK_PATH = os.environ.get('SDK_PATH')
if not SDK_PATH:
  SDK_PATH = guess_sdk_path()

DEV_APPSERVER_PATH = os.path.join(SDK_PATH, 'dev_appserver.py')
PROJECT_PATH = os.getcwd()
SERVER_PATH = os.path.join(PROJECT_PATH, 'server')
AEFS_PATH = os.path.join(PROJECT_PATH, 'aefs.py')
FSTEST_PATH = os.path.join(PROJECT_PATH, 'fstest')

################################################################################

def get_unused_port():
  """Returns an unused port number."""
  port_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  try:
    port_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    port_socket.bind(('localhost', 0))
    return port_socket.getsockname()[1]
  finally:
    port_socket.close()


def run_proc(args):
  """Runs a process with the given argument list, returning a Popen object."""
  try:
    proc = subprocess.Popen(args)
  except:
    logging.exception('Spawning failed: %s', ' '.join(args))
    raise
  logging.info('%s running with pid %d', ' '.join(args), proc.pid)
  return proc


def term_and_wait(proc, send_term=True, log_errors=True):
  """Waits for a process to die after sending it the TERM signal.

  Args:
    proc: A subprocess.Popen instance.
    send_term: If the TERM signal should be sent to the process if it has not
      already terminated.
    log_errors: True if error responses should be logged.

  Returns:
    The exit code of the process.
  """
  result = proc.poll()
  if result is None:
    if send_term:
      logging.info('Sending TERM signal to pid %d', proc.pid)
      os.kill(proc.pid, signal.SIGTERM)
    result = proc.wait()
  if log_errors and result not in (0, -signal.SIGTERM):
    logging.error('%d exited with %d', proc.pid, result)
  return result


class AeFsTestBase(unittest.TestCase):

  def setUp(self):
    """Mounts the fuse directory, starts the backend server."""
    logging.info('**** STARTING: %s', self.id())
    self.port = get_unused_port()
    self.mount = tempfile.mkdtemp(prefix='aefs_test-')

    self.datastore_path = tempfile.mktemp(prefix='aefs_test-')
    self.blobstore_path = tempfile.mktemp(prefix='aefs_test-')

    self.aefs_proc = None
    self.server_proc = run_proc([
        DEV_APPSERVER_PATH,
        '-p', str(self.port),
        '--require_indexes',
        '--datastore_path', self.datastore_path,
        '--blobstore_path', self.blobstore_path,
        SERVER_PATH])

    # Wait for the server socket to open.
    start = time.time()
    connect_exception = None
    while (time.time() - start < 20):
      connect_exception = None
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      sock.settimeout(10)
      try:
        sock.connect(('localhost', self.port))
      except socket.error, e:
        connect_exception = e
        time.sleep(1)
      else:
        sock.close()
        break
    if connect_exception is not None:
      self.fail('Could not connect to local server. %s: %s' %
                (e.__class__.__name__, e))

    self.aefs_proc = run_proc(
        [AEFS_PATH, 'localhost:%d' % self.port, self.mount])
    # Wait for the mount operation to complete.
    parent_device = os.stat(os.path.dirname(self.mount))[stat.ST_DEV]
    start = time.time()
    while True:
      if (time.time() - start) > 5:
        self.fail('aefs.py took too long to mount')
      child_device = os.stat(self.mount)[stat.ST_DEV]
      if child_device != parent_device:
        break
      time.sleep(1)

  def tearDown(self):
    """Unmounts the fuse directory."""
    logging.info('**** FINISHING: %s', self.id())
    term_and_wait(run_proc(['umount', self.mount]), send_term=False)
    if self.aefs_proc: term_and_wait(self.aefs_proc)
    if self.server_proc: term_and_wait(self.server_proc)

  def assertEqualMembers(self, expected, result):
    """Asserts two sortable iterables have equal members."""
    self.assertEquals(sorted(expected), sorted(result))


################################################################################

class AeFsTest(AeFsTestBase):
  """Tests specific to the aefs implementation."""

  def testDirectories(self):
    """Runs directory tests."""
    # Create/list directories and subdirectories.
    self.assertEquals([], os.listdir(self.mount))
    os.mkdir(os.path.join(self.mount, 'foobar'))
    os.mkdir(os.path.join(self.mount, 'meep'))
    os.mkdir(os.path.join(self.mount, 'meep', 'below'))
    os.mkdir(os.path.join(self.mount, 'meep', 'that is'))
    os.mkdir(os.path.join(self.mount, 'stuffo'))
    os.mkdir(os.path.join(self.mount, 'foobar', 'another'))
    os.mkdir(os.path.join(self.mount, 'foobar', 'another', 'deeper'))
    os.mkdir(os.path.join(self.mount, 'foobar', 'okay then'))
    os.mkdir(os.path.join(self.mount, 'foobar', 'some sub directories'))
    self.assertEqualMembers(
        ['foobar', 'meep', 'stuffo'],
        os.listdir(self.mount))

  def testOverlappingNames(self):
    """Tests when two files have the same name in different directories.

    This verifies that hte Inode's base_path and dir_path are being treated
    as tuples and not simple query matches.
    """
    self.assertEquals([], os.listdir(self.mount))
    foobar_dir = os.path.join(self.mount, 'foobar')

    os.mkdir(foobar_dir)
    # Write the bad file first so its key name is lexigraphically first,
    # and thus it will match the query first. This would expose any behavior
    # where we do not traverse the tuples to find an exact path match.
    open(os.path.join(foobar_dir, 'not-matching.txt'), 'w').write('wrong')
    open(os.path.join(foobar_dir, 'stuff.txt'), 'w').write('correct')

    os.mkdir(os.path.join(self.mount, 'meep'))
    meep_dir = os.path.join(self.mount, 'meep')
    os.link(os.path.join(foobar_dir, 'not-matching.txt'),
            os.path.join(meep_dir, 'stuff.txt'))

    result = open(os.path.join(self.mount, 'foobar', 'stuff.txt')).read()
    self.assertEquals('correct', result)

  # TODO: Write more tests for aefs specific edge cases like:
  # - Linking an inode that'd been deleted can resurrect the blob

################################################################################

def generate_fstest_case():
  """Generates a TestCase that runs the fstest files as unittests."""
  containing_dir = os.path.join(FSTEST_PATH, 'tests')
  all_dirs = [d for d in os.listdir(containing_dir)
              if os.path.isdir(os.path.join(containing_dir, d))
              and not d.startswith('.')]

  def generate_test(test_dir):
    def test(self):
      self.assertTrue(os.path.exists(os.path.join(FSTEST_PATH, 'fstest')),
                      'Must build "fstest" binary before running tests.')
      before_dir = os.getcwd()
      os.chdir(self.mount)
      try:
        all_tests = [os.path.join(test_dir, rel_path)
                     for rel_path in os.listdir(test_dir)
                     if rel_path.endswith('.t')]
        failed_tests = []
        successful_tests = []
        for test_path in all_tests:
          proc = subprocess.Popen(['prove', '-v', test_path])
          start = time.time()
          while (time.time() - start) < 10:
            if proc.poll() is None:
              time.sleep(1)
            else:
              break
          else:
            logging.error('Test timed out. Sending KILL to PID = %d', proc.pid)
            os.kill(proc.pid, signal.SIGKILL)

          result = proc.poll()
          if result != 0:
            failed_tests.append(test_path)
          else:
            successful_tests.append(test_path)

        if failed_tests:
          self.fail('%d/%d tests failed: %s\nPassing: %s' %
                    (len(failed_tests), len(all_dirs),
                     ', '.join(failed_tests), ', '.join(successful_tests)))
      finally:
        os.chdir(before_dir)
    return test

  class MetaFsTestCase(type):
    def __new__(cls, name, bases, dict):
      for test_path in all_dirs:
        method_name = 'test_' + re.sub('[\.-/]', '_', test_path)
        func = generate_test(os.path.join(containing_dir, test_path))
        func.func_name = method_name
        dict[method_name] = func
      return type(name, bases, dict)

  class FsTestCase(AeFsTestBase):
    __metaclass__ = MetaFsTestCase

  return FsTestCase


FsTest = generate_fstest_case()

################################################################################

if __name__ == '__main__':
  sys.path.insert(0, SDK_PATH)
  unittest.main()
